package com.spi.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Map;

@RocketMQMessageListener(
        topic = "${rocketmq.consumer.group2.topic}", //主题
        consumerGroup = "${rocketmq.consumer.group2.consumer-group}", //消费者组
        maxReconsumeTimes = 1, //最大重试次数
        messageModel = MessageModel.CLUSTERING, //消费模式 集群
        //messageModel = MessageModel.BROADCASTING, //消费模式 广播
        consumeMode = ConsumeMode.ORDERLY //顺序消费
        //consumeMode = ConsumeMode.CONCURRENTLY //并发消费
)
@Component
@Slf4j
public class ConsumerGroup2Listener implements RocketMQListener<MessageExt>,RocketMQPushConsumerLifecycleListener {
    @Override
    public void onMessage(MessageExt messageExt) {
        //log.info("当前线程名称：" + Thread.currentThread().getName());
        //System.out.println(messageExt);
        byte[] bytes = messageExt.getBody();
        String content = new String(bytes, StandardCharsets.UTF_8);
        Map<String, Object> stringObjectMap = JSON.parseObject(content, new TypeReference<Map<String, Object>>() {});
//        System.out.println("消息内容为：" + stringObjectMap.get("content"));
//        System.out.println("消息id为：" + stringObjectMap.get("msgid"));
        log.info("消息内容为：{}", stringObjectMap.get("content") + "===消息id为：" + stringObjectMap.get("msgid"));
        //log.info("消息id为：{}", stringObjectMap.get("msgid"));
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.setInstanceName("instance1");
    }
}
